package com.codejiwei.flink.table;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * author: codejiwei
 * date: 2023/8/8
 * desc: flink table api stream 2 table
 **/
public class Flink_Table_API_Stream2Table_02 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        //create dataStream
        DataStreamSource<Row> sourceDS = env.fromElements(
                Row.of("Tom", 12),
                Row.of("Jerry", 22),
                Row.of("Tom", 18)
        );

        //TODO dataStream to table
        Table table = tEnv.fromDataStream(sourceDS).as("name", "score");

        tEnv.createTemporaryView("SourceTable", table);
        Table tableResult = tEnv.sqlQuery("select name, SUM(score) from SourceTable group by name");

        DataStream<Row> rowDataStream = tEnv.toChangelogStream(tableResult);

        rowDataStream.print();

        env.execute();


    }
}
